package cc.verywell.pureblock.msg.nfs.idx.extraDownload;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.OverlappingFileLockException;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.security.DigestException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

import cc.verywell.pureblock.func.node.KeyShell;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.msg.nfs.idx.extraDownload.DloTargetCheck.TargetCheck;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadingMonitor;
import cc.verywell.pureblock.msg.nfs.idx.extraUpload.UploadingWatch;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.DataLink;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.DataLinkWrite;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.Sign;
import cc.verywell.pureblock.msg.nfs.strc.NfsStructureV1.Sub;
import cc.verywell.pureblock.net.AioProvider;

/**
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月4日 下午6:18:39
*/
public class DownloadingMonitor implements BiConsumer<KeyShell,byte[]>,Closeable{
	private DloLog log;
	private DloTarget target;
	private DloMap map ;
	private String absolutePath;
	private Path logPath;
	private DownloadManager manager;
	public DownloadingMonitor(String absolutePath,Path logPath,DloLog log, DloTarget target, DloMap map, DownloadManager manager) {
		this.absolutePath = absolutePath;
		this.logPath = logPath;
		this.log = log;
		this.target = target;
		this.map = map;
		this.manager = manager;
		
		this.map.joinWatch(this);
	}
	
	final private AtomicBoolean closed = new AtomicBoolean(false);
	@Override
	public void close() {
		if(closed.compareAndSet(false, true)) {
			//Pblog.DEBUG("before close");
			synchronized (this) {
				//Pblog.DEBUG("into close");
				if(target!=null)
					try {
						target.close();
					} catch (IOException e) {
						if(Pblog.DEBUG)Pblog.DEBUG(e);
					} finally {
						//target = null;
					}
				if(log!=null)
					try {
						log.close();
					} catch (IOException e) {
						if(Pblog.DEBUG)Pblog.DEBUG(e);
					} finally {
						//log = null;
					}
				Set<KeyShell> downloading = map.getLoadingList();
				if(downloading!=null && !downloading.isEmpty()) {
					manager.out.cancelRemoteDownload(this, downloading.toArray(new KeyShell[downloading.size()]));
				}
				if(uploadingMonitor!=null)uploadingMonitor.close();
				//Pblog.DEBUG("end close");
			}
			manager.removeMonitorbyDownloadingMonitor(this);
			//Pblog.DEBUG("out close");

		}
	}
	
	private UploadingMonitor uploadingMonitor;
	private UploadingMonitor initUpMon() {
		UploadingMonitor m = this.uploadingMonitor;
		if(m == null) {
			synchronized (this) {
				if(closed.get())return null;
				if(this.uploadingMonitor==null) {
					this.uploadingMonitor = manager.out.getUploadMonitor(absolutePath);
				}
				m = this.uploadingMonitor;
			}
		}
		return m;
	}
	void fireNewSeedReady(TargetCheck check,long totalSize) {
		UploadingMonitor m = initUpMon();
		if(closed.get() || m==null)return;
		m.newDataReady(check.key, totalSize, check.pos.pos, check.pos.len);
	}
	private void tryHoldInternal(KeyShell key,byte[] payload) {
		if(target.isEmptyPath)return;
		UploadingMonitor m = initUpMon();
		if(closed.get() || m==null)return;
		m.tryHoldInternal(key, payload);
	}
	
	void fireFinished() {
		//此时log和target已经关闭
		//通知外部
		manager.fireFinishbyDownloadingMonitor(this);
	}
	public boolean isFinished() {
		return map.isFinished();
	}

	

	public void wantList(KeyShell... keys) {
		if(closed.get())return;
		LinkedList<KeyShell> al = new LinkedList<>(Arrays.asList(keys));
		Iterator<KeyShell> it = al.iterator();
		while(it.hasNext()) {
			KeyShell key = it.next();
			byte[] payload = manager.out.getLocalData(key);
			if(payload!=null) {
				it.remove();
				try {
					blockArrival(key, payload, true);
				} catch (DigestException | IOException | InterruptedException e) {
					if(Pblog.DEBUG)Pblog.DEBUG(e);
					close();
					return;
				}
				if(closed.get())return;
			}
		}
		if(!al.isEmpty()) {
			manager.out.getRemoteData(this, al.toArray(new KeyShell[al.size()]));
		}
	}
	@Override
	public void accept(KeyShell t, byte[] u) {
		
		if(closed.get())return;
		try {
			if(u!=null)blockArrival(t, u, false);
			else wantList(t);
		} catch (DigestException | IOException | InterruptedException e) {
			if(Pblog.DEBUG)Pblog.DEBUG(e);
			close();
		}
	}
	
	private void blockArrival(KeyShell key,byte[] payload,boolean isLocal) throws IOException, DigestException, InterruptedException {
//		HashMap<KeyShell,Download_WAIT> map = new HashMap<KeyShell,Download_WAIT>();
//		MD5 md5 = new MD5();
		
		NfsStructureV1 s = new NfsStructureV1(payload);
		final DloMapWait w = map.get(key, s);
		if(w==null) {
			if(Pblog.DEBUG)Pblog.DEBUG("get not need bag! ",key.key);
			return; //没有w就证明这里不需要;
		}
		try {
			boolean isPuredata;
			ArrayList<KeyShell> waitList = null;
			WriteDataPack joindata;
			synchronized (log) {
				DltoolIterator<KeyShell> key_history = null;
				long topused=0,endused=0;
				if(s.data!=null) {
//					if(!s.isPuredata()) {
//						Pblog.DEBUG("not puredata located");
//					}
					int len = s.data.getLength();
					//先搞定目前的数据写入
					DltoolIterator<DltoolPosLen> pos = w.dataonly_list;
					while(pos!=null) {
						if(pos.t.len!=len)throw new IOException("err block data waiting size !");
						pos = pos.prv;
					}
					DltoolIterator<DltoolPosLen> exportPos = w.dataonly_list;
					pos = w.structure_list;
					while(pos!=null) {
						if(pos.t.len-topused-endused<len)throw new IOException("err block structure waiting size !");
						exportPos = DltoolIterator.toNew(exportPos, new DltoolPosLen(pos.t.pos+pos.t.len-len, len));
						pos = pos.prv;
					}			
					endused = len;
					if(exportPos==null)throw new IOException("err block data no place need write !");
					ByteBuffer outbf = s.data.getByteBuffer();
					byte[] mdHash = DloTargetCheck.hash.get().hash(outbf);
					//不可outbf.position(0);
					log.logWrite_get(key, mdHash, exportPos);
					joindata = new WriteDataPack(outbf, exportPos, key, mdHash, isPuredata = s.isPuredata());//将要写入的内容交给异步去做;
				}else { //纯结构包要进行迭代检查
					if(w.dataonly_list!=null)throw new IOException("err block data waiting no have!");//！需要数据却没有
					DltoolIterator<KeyShell> kh = w.key_history;
					while(kh!=null) {
						if(kh.t.equals(key))throw new IOException("err block structure iteratored !");//！迭die代检查发现有重复
						kh = kh.prv;
					}
					key_history = DltoolIterator.toNew(w.key_history, key);
					log.logWrite_get(key, null, null);
					isPuredata = false;
					joindata = null;
				}
				//下级datas分析及处理
				if(s.datas!=null){
					waitList = new ArrayList<>(s.datas.length);
					for (DataLink d : s.datas) {
						KeyShell dkey = new KeyShell(d.getHash());
						DltoolIterator<DltoolPosLen> exportPos = null;
						DltoolIterator<DltoolPosLen> pos = w.structure_list;
						while(pos!=null) {
							if(pos.t.len-topused-endused<d.getSize())throw new IOException("err block structure waiting sub size !");//！下面的结构是错的
							exportPos = DltoolIterator.toNew(exportPos, new DltoolPosLen(pos.t.pos+topused, d.getSize()));
							pos = pos.prv;
						}
						topused+=d.getSize();
						log.logWrite_wait(dkey, exportPos);
						if(map.add_structure_req(dkey,key_history,exportPos)) {
							waitList.add(dkey);
						};
					}
				}
				{
					DltoolIterator<DltoolPosLen> pos = w.structure_list;
					while(pos!=null) {
						if(pos.t.len-topused-endused!=0)throw new IOException("err block structure waiting have hole !");//！有数据空缺
						pos = pos.prv;
					}
				}
				///其它数据记录
				log.logWrite_extra(s);
			}
			///与log无关数据在最后执行
			if(joindata!=null) {
				joindata.run();//将要写入的内容交给异步去做;
				joindata = null;
			}
			if(!isPuredata){
				tryHoldInternal(key, payload);
//				if(isLocal)manager.out.holdInternal(this, key);
//				else manager.out.putAndHoldInternal(this, key, payload);
			}
			///新key搜索在此添加，否则会令log记录混乱
			if(waitList!=null && !waitList.isEmpty()) {
				wantList(waitList.toArray(new KeyShell[waitList.size()]));
			}
		}finally {
			map.checkFinish(w); // 告知w包已经使用完毕;
		}
	}
	private class WriteDataPack{
		private ByteBuffer outbf;
		private DltoolIterator<DltoolPosLen> exportPos;
		private KeyShell key;
		private byte[] mdHash;
		private boolean isPuredata;
		
		public WriteDataPack(ByteBuffer outbf, DltoolIterator<DltoolPosLen> exportPos, KeyShell key, byte[] mdHash,	boolean isPuredata) {
			this.outbf = outbf;
			this.exportPos = exportPos;
			this.key = key;
			this.mdHash = mdHash;
			this.isPuredata = isPuredata;
		}

		public void run() throws IOException, DigestException, InterruptedException {
			// TODO Auto-generated method stub
			target.joindata(outbf, exportPos, key, mdHash, isPuredata);
		}
	}
	
	
	public List<Sub> getAlias() throws IOException{
		ArrayList<Long> logpos = this.log.posAlias;
		if(logpos==null)return null;
		if(logpos.isEmpty())return null;
		LinkedList<Sub> result = new LinkedList<>();
		getSub(logPath, logpos, (body ,r)->{
			int alsize = 0xFFFF & body.getShort();
			while(body.hasRemaining()) {
				int len = 0xFFFF & body.getShort();
				byte[] bs = new byte[len];
				body.get(bs);
				r.add(new NfsStructureV1.SubWrite(bs));
				alsize--;
			}
			if(alsize!=0)Pblog.FS("readLog file getAlias lack err: ",alsize," at ",logPath);
		}, result);
		return result;
	}
	public List<Sign> getSigns() throws IOException{
		ArrayList<Long> logpos = this.log.posSign;
		if(logpos==null)return null;
		if(logpos.isEmpty())return null;
		LinkedList<Sign> result = new LinkedList<>();
		getSub(logPath, logpos, (body ,r)->{
			int alsize = 0xFFFF & body.getShort();
			while(body.hasRemaining()) {
				int len = 0xFFFF & body.getShort();
				byte[] bs = new byte[len];
				body.get(bs);
				Sub pubKey = new NfsStructureV1.SubWrite(bs);
				len = 0xFFFF & body.getShort();
				bs = new byte[len];
				body.get(bs);
				Sub sign = new NfsStructureV1.SubWrite(bs);
				r.add(new Sign(pubKey, sign));
				alsize--;
			}
			if(alsize!=0)Pblog.FS("readLog file getSigns lack err: ",alsize," at ",logPath);
		}, result);
		return result;
	}
	public List<DataLink> getExtras() throws IOException{
		ArrayList<Long> logpos = this.log.posExtra;
		if(logpos==null)return null;
		if(logpos.isEmpty())return null;
		LinkedList<DataLink> result = new LinkedList<>();
		getSub(logPath, logpos, (body ,r)->{
			int alsize = 0xFFFF & body.getShort();
			while(body.hasRemaining()) {
				long size = body.getLong();
				int len = 0xFFFF & body.getShort();
				byte[] bs = new byte[len];
				body.get(bs);
				r.add(new DataLinkWrite(bs, size));
				alsize--;
			}
			if(alsize!=0)Pblog.FS("readLog file getSigns lack err: ",alsize," at ",logPath);
		}, result);
		return result;
	}
	
	
	static public <T> void  getSub(Path logPath,ArrayList<Long> logpos,BiConsumer<ByteBuffer,T> func,T t) throws IOException {
		ByteBuffer head = ByteBuffer.allocate(8);
		ByteBuffer body = null;
		try( SeekableByteChannel channel = Files.newByteChannel(logPath,StandardOpenOption.READ) ) {
			for (Long pos : logpos) {
				SeekableByteChannel c = channel.position(pos);
				head.clear();
				if(head.capacity()!=c.read(head))throw new IOException("readLog file head err: "+head.capacity()+" => "+head.position()+" at "+logPath+":"+pos);
				head.flip();
				long size = head.getLong();
				if(size<=0 || size>AioProvider.FILE_MAX_SIZE) throw new IllegalArgumentException("readLog size "+size+" too big at "+logPath+":"+pos);
				if(body!=null && body.capacity()>=size) {
					body.clear();
				}else {
					int newSize = (int)(size*1.2);
					if(newSize<0 || newSize>AioProvider.FILE_MAX_SIZE)
						newSize = AioProvider.FILE_MAX_SIZE;
					body = ByteBuffer.allocate(newSize);
				}
				body.limit((int)size);
				if(body.limit()!=c.read(body))throw new IOException("readLog file body err: "+body.limit()+" => "+body.position()+" at "+logPath+":"+pos);
				body.flip();
				func.accept(body, t);
				if(body.hasRemaining())throw new IOException("readLog file body lack err: "+body.limit()+" => "+body.position()+" at "+logPath+":"+pos);
			}
		}
	}
	


}
